package com.example.mq.mqserver.datacenter;

import com.example.mq.common.BinaryTool;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;

import java.io.*;
import java.util.LinkedList;
import java.util.Scanner;

//针对存储消息的文件进行管理
public class MessageFileManager {
    //定义内部类表示该队列的统计信息
    static public class Stat {
        public int totalCount;//总数量
        public int validCount;//有效消息数量
    }

    public void init(){
        //暂时不需要初始化
    }

    //约定消息文件所在的目录和文件名
    //获取指定队列对应的消息文件所在路径
    private String getQueueDir(String queueName) {
        return "./data/" + queueName;
    }

    //这个方法用来获取该队列的消息数据文件路径
    private String getQueueDataPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data.txt";
    }

    //这个方法用来获取该队列的消息统计文件路径
    private String getQueueStatPath(String queueName) {
        return getQueueDir(queueName) + "/queue_stat.txt";
    }

    //读统计文件
    private Stat readStat(String queueName) {
        Stat stat = new Stat();
        try (InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))) {
            Scanner scanner = new Scanner(inputStream);
            stat.totalCount = scanner.nextInt();
            stat.validCount = scanner.nextInt();
            return stat;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    //写统计文件
    private void writeStat(String queueName, Stat stat) {
        //使用printWrite
        //outputstream打开文件默认情况下啊 会把源文件清空 此时相当于旧文件覆盖新文件
        try (OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))) {
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.write(stat.totalCount + "\t" + stat.validCount);
            printWriter.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    public void createQueueFiles(String queueName) throws IOException {
        // 1.创建⽬录指定队列的⽬录

        File baseDir = new File(getQueueDir(queueName));
        if (!baseDir.exists()) {
            boolean ok = baseDir.mkdirs();
            if (!ok) {
                throw new IOException(" 创建⽬录失败 ! baseDir=" + baseDir.getAbsolutePath());
            }
        }
        // 2.创建队列数据⽂件

        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            boolean ok = queueDataFile.createNewFile();
            if (!ok) {
                throw new IOException(" 创建⽂件失败 ! queueDataFile=" + queueDataFile.getAbsolutePath());
            }
        }
        // 3.创建队列统计⽂件

        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            boolean ok = queueStatFile.createNewFile();
            if (!ok) {
                throw new IOException(" 创建⽂件失败 ! queueStatFile=" + queueStatFile.getAbsolutePath());
            }
        }
        // 4.给队列统计⽂件写⼊初始数据

        Stat stat = new Stat();
        stat.totalCount = 0;
        stat.validCount = 0;
        writeStat(queueName, stat);
    }

    //删除队列的目录和文件
    //队列也是可以删除的 当队列删除之后 对应的消息文件也会被删除
    public void destroyQueueFiles(String queueName) throws IOException {
        //先删除里面的文件 在删除目录
        File queueDataFile = new File(getQueueDataPath(queueName));
        boolean ok1 = queueDataFile.delete();
        File queueStatFile = new File(getQueueStatPath(queueName));
        boolean ok2 = queueStatFile.delete();
        File baseDir = new File(getQueueDir(queueName));
        boolean ok3 = baseDir.delete();
        if (!ok1 || !ok2 || !ok3) {
            //其中有任意一个失败 都算整体删除失败
            throw new IOException("删除队列目录和文件失败 baseDir=" + baseDir.getAbsolutePath());
        }
    }

    //检查队列的目录和文件是否存在
    public boolean checkFilesExists(String queueName) {
        //判定队列的数据文件 和统计文件是否都存在
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            return false;
        }
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            return false;
        }
        return true;
    }

    //把一个新的消息 放到队列对应的文件中
    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
        //检查档期那要写入的队列对应的文件是否存在
        if (!checkFilesExists(queue.getName())) {
            throw new MqException("[MessageFileManager] 队列对应文件不存在 queueName = " + queue.getName());
        }
        //把message对象进行序列化 转化成二进制字节数组
        byte[] messageBinary = BinaryTool.toBytes(message);

        synchronized (queue) {
            //获取到当前队列数据文件的长度 用来计算message的 offsetBeg和offsetEnd
            //offsetEnd 就是当前文件长度 + 4 + message自身长度
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
            message.setOffsetBeg(queueDataFile.length() + 4);
            message.setOffsetEnd(queueDataFile.length() + 4 + messageBinary.length);
            //写消息到文件中 追加到数据文件末尾 而不是覆盖
            try (OutputStream outputStream = new FileOutputStream(queueDataFile, true)) {
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    //接下来先写当前消息的长度
                    //这个操作就是写入四个字节了
                    dataOutputStream.writeInt(messageBinary.length);
                    //写入消息本体
                    dataOutputStream.write(messageBinary);
                }
            }
            //更新消息统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount += 1;
            stat.validCount += 1;
            writeStat(queue.getName(), stat);
        }
    }

    //删除消息
    //逻辑删除 将isValid属性设置成0
    //先把文件中的数据还原为message对象 isValid改成0 在重新写回去
    //此处message必须包含有效的offsetBeg 和OffsetEnd
    public void deleteMessage(MSGQueue queue, Message message) throws IOException, ClassNotFoundException {
        //随机访问 在文件的指定位置读写
        //seek方法移动光标
        synchronized (queue) {
            try (RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()), "rw")) {
                //先从文件中读取相应的message数据
                byte[] bufferSrc = new byte[(int) (message.getOffsetEnd() - message.getOffsetBeg())];
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.read(bufferSrc);
                //把二进制数据转化为Message对象
                Message diskMessage = (Message) BinaryTool.fromBytes(bufferSrc);
                //设置isValid
                diskMessage.setIsValid((byte) 0x0);
                //重新写入
                byte[] bufferDest = BinaryTool.toBytes(diskMessage);
                //此时重新 seek 虽然上面已经seek过 但是进行了读操作 导致文件光标向后移动到下一个消息的位置
                //因此需要重新seek 调整光标
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.write(bufferDest);
            }
            //更新统计文件
            Stat stat = readStat(queue.getName());
            if ((stat.validCount > 0)) {
                stat.validCount -= 1;
            }
            writeStat(queue.getName(), stat);
        }
    }

    //从文件中获取所有的消息内容 加载到内存中(一个链表中)
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, MqException, ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();
        try (InputStream inputStream = new FileInputStream(getQueueDataPath(queueName))) {
            try (DataInputStream dataInputStream = new DataInputStream(inputStream)) {
                //模拟光标位置
                long currentOffset = 0;
                //循环读取消息
                while (true) {
                    //读取当前消息
                    //这里可能会读到文件末尾 readInt方法到达末尾会抛出EOFException异常
                    int messageSize = dataInputStream.readInt();
                    //根据消息长度 读取消息
                    byte[] buffer = new byte[messageSize];
                    int actualSize = dataInputStream.read(buffer);
                    if (messageSize != actualSize) {
                        throw new MqException("[MessageFileManager] 文件格式错误 queueName = " + queueName);
                    }
                    //把读到的二进制数据 反序列化为Message对象
                    Message message = (Message) BinaryTool.fromBytes(buffer);
                    //判断这个小时是否为无效对象
                    if (message.getIsValid() != 0x1) {
                        //无效数据直接跳过
                        currentOffset += (4 + messageSize);
                        continue;
                    }
                    //有效数据 则需要把这个数据加入到链表中 加入前要填写offsetBeg offsetEnd
                    //进行计算时 需要知道当前光标的位置
                    //此时手动计算下标
                    message.setOffsetBeg(currentOffset + 4);
                    message.setOffsetEnd(currentOffset + 4 + messageSize);
                    currentOffset += (4 + messageSize);
                    //添加到链表中
                    messages.add(message);
                }
            }catch (EOFException e){
                //此时不是真正处理异常 而是处理正常业务逻辑
                System.out.println("[MessageFileManager] 恢复Message数据完成");
            }
        }
        return messages;
    }

    //检查当前是否要针对该队列的消息数据文件进行GC
    public boolean checkGC(String queueName){
        //判定是否要GC 是根据总消息数和有效消息数 这两个值 是在消息统计文件中的
        Stat stat = readStat(queueName);
        if(stat.totalCount > 2000 && (double)stat.validCount /(double) stat.totalCount < 0.5){
            return true;
        }
        return false;
    }

    private String getQueueDataNewPath(String queueName){
        return getQueueDir(queueName) + "/queue_data_new.txt";
    }

    //垃圾回收操作
    //使用复制算法
    //创建新文件 名字为queue_data_new.txt
    //把之前的消息数据文件中有效的消息读取出来 写到新文件中
    //删除旧的消息 把文件名改回queue_data.txt
    //同时要记得更新消息统计文件
    public void gc(MSGQueue queue) throws MqException, IOException, ClassNotFoundException {
        //进行gc时 其他线程不能对文件进行修改
        synchronized (queue){
            //由于gc操作比较耗时 此处统计消耗时间
            long gcBeg = System.currentTimeMillis();
            //创建一个新文件
            File queueDataNewFile = new File(getQueueDataNewPath(queue.getName()));
            if(queueDataNewFile.exists()){
                //正常情况下 这个文件不应该存在 说明上一次gc一半 程序意外结束
                throw  new MqException("[MessageFileManager] gc 时发现该队列的queue_data_new 已经存在");
            }
            boolean ok = queueDataNewFile.createNewFile();
            if(!ok){
                throw new MqException("[MessageFileManager] 创建文件失败 queueDataNewFile =" + queueDataNewFile.getAbsolutePath() );
            }
            //从旧文件中获取出所有有效消息
            LinkedList<Message> messages = loadAllMessageFromQueue(queue.getName());
            //把有效消息 写入到新文件
            try(OutputStream outputStream = new FileOutputStream(queueDataNewFile,true)){
                try(DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){
                    for(Message message : messages){
                        byte[] buffer = BinaryTool.toBytes(message);
                        dataOutputStream.writeInt(buffer.length);
                        dataOutputStream.write(buffer);
                    }
                }
            }
            //删除旧文件 把新文件重命名
            File queueDataOldFile = new File(getQueueDataPath(queue.getName()));
            ok = queueDataOldFile.delete();
            if(!ok){
                throw new MqException("[MessageFileManager] 删除旧数据失败 queueDataOldFile" + queueDataOldFile.getAbsolutePath());
            }
            //重命名文件
            ok = queueDataNewFile.renameTo(queueDataOldFile);
            if(!ok){
                throw new MqException("[MessageFileManager] 文件重命名失败 queueDataNewFile" + queueDataNewFile.getAbsolutePath() +
                        ", queueDataOldFile = " + queueDataOldFile.getAbsolutePath());
            }

            //更新统计文件
            Stat stat = new Stat();
            stat.totalCount = messages.size();
            stat.validCount = messages.size();
            writeStat(queue.getName(),stat);
            long gcEnd = System.currentTimeMillis();
            System.out.println("[MessageFileManager] gc 执行完毕 queueName = " + queue.getName() +
                    "time: " + (gcEnd - gcBeg) + " ms");
        }
    }

}
